/*
 * Copyright 2019 WeBank
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.webank.wedatasphere.linkis.bml.service.impl;

import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.webank.wedatasphere.linkis.bml.Entity.ResourceVersion;
import com.webank.wedatasphere.linkis.bml.Entity.Version;
import com.webank.wedatasphere.linkis.bml.common.Constant;
import com.webank.wedatasphere.linkis.bml.common.ResourceHelper;
import com.webank.wedatasphere.linkis.bml.common.ResourceHelperFactory;
import com.webank.wedatasphere.linkis.bml.dao.VersionDao;
import com.webank.wedatasphere.linkis.bml.service.ResourceService;
import com.webank.wedatasphere.linkis.bml.service.VersionService;
import com.webank.wedatasphere.linkis.common.io.Fs;
import com.webank.wedatasphere.linkis.common.io.FsPath;
import com.webank.wedatasphere.linkis.storage.FSFactory;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.glassfish.jersey.media.multipart.FormDataBodyPart;
import org.glassfish.jersey.media.multipart.FormDataContentDisposition;
import org.glassfish.jersey.media.multipart.FormDataMultiPart;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by cooperyang on 2019/5/17.
 */
@Service
public class VersionServiceImpl implements VersionService {

    private static final Logger logger = LoggerFactory.getLogger(VersionServiceImpl.class);

    @Autowired
    private VersionDao versionDao;

    @Autowired
    private ResourceService resourceService;

    @Override
    public Version getVersion(String resourceId, String version) {
        return versionDao.getVersion(resourceId, version);
    }



    @Override
    public List<ResourceVersion> getResourcesVersions(Map paramMap) {
        return versionDao.getResourcesVersions(paramMap);
    }

    @Override
    public void deleteResourceVersion(String resourceId, String version) {
        versionDao.deleteVersion(resourceId, version);
    }

    @Override
    public void deleteResourceVersions(String resourceId) {
    }

    @Override
    public void deleteResourcesVersions(List<String> resourceIds, List<String> versions) {

    }

    @Override
    public String updateVersion(String resourceId, String user, FormDataMultiPart formDataMultiPart,
                                Map<String, Object> params)throws Exception{
        ResourceHelper resourceHelper = ResourceHelperFactory.getResourceHelper();
        FormDataBodyPart file = formDataMultiPart.getField("file");
        InputStream inputStream = file.getValueAs(InputStream.class);
        final String resourceIdLock = resourceId.intern();
        FormDataContentDisposition fileDetail = file.getFormDataContentDisposition();
        String fileName = new String(fileDetail.getFileName().getBytes("ISO8859-1"), "UTF-8");
        //获取资源的path
        String path = versionDao.getResourcePath(resourceId);
        String newVersion;
        //上传资源前，需要对resourceId这个字符串的intern进行加锁，这样所有需要更新该资源的用户都会同步
        synchronized (resourceIdLock){
            //资源上传到hdfs
            StringBuilder stringBuilder = new StringBuilder();
            long size = resourceHelper.upload(path, user, inputStream, stringBuilder);
            String md5String = stringBuilder.toString();
            String clientIp = params.get("clientIp").toString();
            //生成新的version
            String lastVersion = versionDao.getNewestVersion(resourceId);
            newVersion = params.get("newVersion").toString();
            long startByte = versionDao.getEndByte(resourceId, lastVersion) + 1;
            //更新resource_version表
            ResourceVersion resourceVersion = ResourceVersion.createNewResourceVersion(resourceId, path,
                    md5String, clientIp, size, newVersion, startByte);
            versionDao.insertNewVersion(resourceVersion);
        }
        return newVersion;
    }

    private String generateNewVersion(String version){
        int next = Integer.parseInt(version.substring(1, version.length())) + 1;
        return Constant.VERSION_PREFIX + String.format(Constant.VERSION_FORMAT, next);
    }

    @Override
    public String getNewestVersion(String resourceId) {
        return versionDao.getNewestVersion(resourceId);
    }

    @Override
    public boolean downloadResource(String user, String resourceId, String version, OutputStream outputStream,
                                    Map<String, Object> properties)throws IOException {
        //1获取resourceId 和 version对应的资源所在的路径
        //2获取的startByte和EndByte
        //3使用storage获取输入流
        ResourceVersion resourceVersion = versionDao.findResourceVersion(resourceId, version);
        long startByte = resourceVersion.getStartByte();
        long endByte = resourceVersion.getEndByte();
        String path = resourceVersion.getResource();
        Fs fileSystem = FSFactory.getFsByProxyUser(new FsPath(path), user);
        fileSystem.init(new HashMap<String, String>());
        InputStream inputStream = fileSystem.read(new FsPath(path));
        inputStream.skip(startByte - 1);
        logger.info("{} 下载资源 {} inputStream skipped {} bytes", user, resourceId, (startByte - 1));
        byte[] buffer = new byte[1024];
        long size = endByte - startByte + 1;
        int left = (int) size;
        try {
            while(left > 0) {
                int readed = inputStream.read(buffer);
                int useful = Math.min(readed, left);
                if(useful < 0){
                    break;
                }
                left -= useful;
                byte[] bytes = new byte[useful];
                for (int i = 0; i <useful ; i++) {
                    bytes[i] = buffer[i];
                }
                outputStream.write(bytes);
            }
        }finally {
            //int size = IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
            fileSystem.close();
        }
        return size >= 0 ;
    }

    @Override
    public List<Version> getVersions(String resourceId) {
        return versionDao.getVersions(resourceId);
    }
//    @Override
//    public List<Version> getVersions(String resourceId, List<String> versions) {
//        return versionDao.getVersions(resourceId, versions);
//    }

    //分页查询
    public List<Version> selectVersionByPage(int currentPage, int pageSize,String resourceId){
        List<Version> rvList = null;
        if(StringUtils.isNotEmpty(resourceId)){
            PageHelper.startPage(currentPage, pageSize);
            rvList =versionDao.selectVersionByPage(resourceId);
        } else {
            rvList = new ArrayList<Version>();
        }
        PageInfo<Version> pageInfo = new PageInfo<>(rvList);
        return pageInfo.getList();
    }

    @Override
    public List<ResourceVersion> getAllResourcesViaSystem(String system, String user) {
        return versionDao.getAllResourcesViaSystem(system, user);
    }
    @Override
    public List<ResourceVersion> selectResourcesViaSystemByPage(int currentPage, int pageSize,String system, String user){
        List<ResourceVersion> resourceVersions = null;
        if(StringUtils.isNotEmpty(system) || StringUtils.isNotEmpty(user)){
            PageHelper.startPage(currentPage, pageSize);
            resourceVersions = versionDao.selectResourcesViaSystemByPage(system, user);
        } else {
            resourceVersions = new ArrayList<ResourceVersion>();
        }
        PageInfo<ResourceVersion> pageInfo = new PageInfo<>(resourceVersions);
        return pageInfo.getList();
    }

    @Override
    public boolean checkVersion(String resourceId, String version) {
        return versionDao.checkVersion(resourceId, version) == 1;
    }

    @Override
    public boolean canAccess(String resourceId, String version) {
        return versionDao.selectResourceVersionEnbleFlag(resourceId, version) == 1;
    }


}
